package com.roy.minimq.remoting.domain.nettyserver;

import com.roy.minimq.remoting.domain.entity.NettyEvent;
import com.roy.minimq.remoting.domain.entity.RemotingServer;
import com.roy.minimq.remoting.domain.entity.common.Pair;
import com.roy.minimq.remoting.domain.netty.*;
import com.roy.minimq.remoting.service.ChannelEventListener;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author ：楼兰
 * @date ：Created in 2021/8/9
 * @description:
 **/

public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    private ServerBootstrap serverBootstrap;
    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;
    private NettyServerConfig nettyServerConfig;
    private DefaultEventExecutorGroup defaultEventExecutorGroup;
    private ChannelEventListener channelEventListener;
    private NettyEventExecutor nettyEventExecutor;
    private ExecutorService publicExecutor;

    public NettyRemotingServer(NettyServerConfig nettyServerConfig, ChannelEventListener channelEventListener){
        this.channelEventListener = channelEventListener;
        this.nettyEventExecutor = new NettyEventExecutor(channelEventListener);
        this.serverBootstrap = new ServerBootstrap();
        this.bossGroup = new NioEventLoopGroup(1,new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyEPOLLBoss_%d", this.threadIndex.incrementAndGet()));
            }
        });

        this.workerGroup = new NioEventLoopGroup(0, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);
            private int threadTotal = nettyServerConfig.getServerSelectorThreads();

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("NettyServerEPOLLSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
            }
        });
        this.nettyServerConfig = nettyServerConfig;
        this.publicExecutor = Executors.newFixedThreadPool(4, new ThreadFactory() {
            private AtomicInteger threadIndex = new AtomicInteger(0);
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "NettyServerPublicExecutor_" + this.threadIndex.incrementAndGet());
            }
        });
    }
    @Override
    public void start() {

        if (this.channelEventListener != null) {
            this.nettyEventExecutor.start();
        }

        this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
                nettyServerConfig.getServerWorkerThreads(),
                new ThreadFactory() {
                    private AtomicInteger threadIndex = new AtomicInteger(0);

                    @Override
                    public Thread newThread(Runnable r) {
                        return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
                    }
                });
        NettyServerConnectManageHandler nettyServerConnecManageHandler = new NettyServerConnectManageHandler(this);
        NettyServerHandler nettyServerHandler = new NettyServerHandler(this);
        this.serverBootstrap.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG,1024)
                .option(ChannelOption.SO_REUSEADDR,true)
                .childOption(ChannelOption.SO_KEEPALIVE,false)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
                .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
                .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        final ChannelPipeline ch = socketChannel.pipeline();
                        ch.addLast(defaultEventExecutorGroup,
                                new NettyDecoder(),
                                new NettyEncoder(),
                                new IdleStateHandler(0,0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
                                nettyServerConnecManageHandler,
                                nettyServerHandler
                        );
                    }
                });

        try{
            final ChannelFuture future = this.serverBootstrap.bind(nettyServerConfig.getListenPort()).sync();
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if(channelFuture.isDone()){
                        logger.info("remoting server started at port "+ nettyServerConfig.getListenPort());
                    }
                }
            });
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException("the serverbootstrap.bind().sync() interrupted ",e);
        }
    }

    @Override
    public void shutdown() {
        if(null != this.defaultEventExecutorGroup){
            defaultEventExecutorGroup.shutdownGracefully();
        }
        if(null != this.bossGroup){
            bossGroup.shutdownGracefully();
        }
        if(null != this.workerGroup){
            workerGroup.shutdownGracefully();
        }
    }
    public void putNettyEvent(NettyEvent event){
        this.nettyEventExecutor.putNettyEvent(event);
    }

    public ChannelEventListener getChannelEventListener() {
        return channelEventListener;
    }

    @Override
    public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executorService) {
        this.processorTable.put(requestCode,new Pair<>(processor,executorService));

    }

    public void registerDefaultProcessor(NettyRequestProcessor processor, ExecutorService executorService){
        this.defaultRequestProcessor = new Pair<>(processor,executorService);
    }

    @Override
    public Pair<NettyRequestProcessor, ExecutorService> getProcessorPair(int requestCode) {
        return this.processorTable.get(requestCode);
    }
}
